package cn.dglydrpy.study.mapreduce.temperaturetest;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;

public class TempertureReducer extends Reducer<Temperature,Temperature,Temperature, NullWritable> {
    int cnt;
    //组织Map<K,V> K:年-月 V：天,当前年月写出的数据条数
    HashMap<String,String> flagmap = new HashMap<String,String>();

    //创建集合，将当前分区数据放入该集合
    ArrayList<Temperature> list = new ArrayList<Temperature>();
    @Override
    protected void reduce(Temperature key, Iterable<Temperature> values, Reducer<Temperature, Temperature, Temperature, NullWritable>.Context context) throws IOException, InterruptedException {

        //遍历所有数据 放入一个集合中
        Iterator<Temperature> iterator = values.iterator();
        while(iterator.hasNext()){
            Temperature next = iterator.next();
            list.add(next);
        }

        //遍历list集合，来取出相同年月 不同日期的top2 两条数据
        for (Temperature temperature : list) {
            String year = temperature.getYear();
            String month = temperature.getMonth();
            String day = temperature.getDay();

            //第一次处理当前年月数据，直接写出
            if(!flagmap.containsKey(year+"-"+month)){
                cnt = 1;
                context.write(temperature,NullWritable.get());
                flagmap.put(year+"-"+month,day+","+cnt);
            }

            //当前年月数据有多条，判断天是否相同，不相同就输出，相同跳过
            if(flagmap.containsKey(year+"-"+month) && !day.equals(flagmap.get(year+"-"+month).split(",")[0])){
                cnt = Integer.valueOf(flagmap.get(year + "-" + month).split(",")[1]);
                cnt+=1;
                if(cnt == 2){
                    context.write(temperature,NullWritable.get());
                }
                flagmap.put(year+"-"+month,day+","+cnt);
            }


        }

    }
}